package com.we.flink.utils;

import org.apache.flink.api.java.utils.ParameterTool;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;

public class WeCdcPropertyReader {
    private static ParameterTool param;
    public static Logger LOG = LoggerFactory.getLogger(WeCdcPropertyReader.class);

    public static WeCdcPropertyReader init(String propFilePath) {
        try {
            param = getReader(propFilePath);
        } catch (Exception e) {
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
        return new WeCdcPropertyReader();
    }

    public static ParameterTool getReader(String filePath) throws IOException {
        InputStream resourceAsStream =
                WeCdcPropertyReader.class.getClassLoader().getResourceAsStream(filePath);
        return ParameterTool.fromPropertiesFile(resourceAsStream);
    }

    public String getHost() {
        return param.get("host");
    }

    public int getPort() {
        return param.getInt("port");
    }

    public String getUserName() {
        return param.get("username");
    }

    public String getPassword() {
        return param.get("password");
    }

    public String getDatabase() {
        return param.get("database");
    }

    public String getTableName() {
        return param.get("tablename");
    }

    public String getServerId() {
        return param.get("serverid");
    }

    public int getSplitNum() {
        return param.getInt("splitnum");
    }

    public String getKfkSinkAdd() {
        return param.get("kfksinadd");
    }

    public String getKfkTopicPrefix() {
        return param.get("kfktopicprefix");
    }

    public String getKfkTopicSuffix() {
        return param.get("kfktopicsuffix");
    }

    public String getKeySeriaSchemaClm() {
        return param.get("keyseriaschemaclm");
    }

    public String getRocksDBBackendUrl() {
        return param.get("rocksDBBackendURL");
    }

    public static void main(String[] args) {
        WeCdcPropertyReader reader =
                WeCdcPropertyReader.init("cdcpipeline/Hhl01TestSource.properties");

        String tablelist = "";
        String database = reader.getDatabase();
        String tableName = reader.getTableName();

        System.out.println(reader.getHost());
        System.out.println(reader.getPort());
        System.out.println(reader.getUserName());
        System.out.println(reader.getPassword());
        System.out.println(database);
        System.out.println(tableName);
        System.out.println(reader.getSplitNum());
        System.out.println(reader.getKfkSinkAdd());
        System.out.println(reader.getKeySeriaSchemaClm());

        if (tableName.contains(",")) {
            String[] splits = tableName.split(",");
            for (String split : splits) {
                tablelist += database + "." + split + ",";
            }
            tablelist = tablelist.substring(0, tablelist.length() - 1);
        } else {
            tablelist += database + "." + tableName;
        }

        System.out.println(tablelist);

        //        String tablelist = "";
        //        String database = reader.getDatabase();
        //        String tableName = reader.getTableName();
        //        int splitNum = reader.getSplitNum();
        //        if(splitNum > 1) {
        //            for (int i = 0; i < splitNum; i++) {
        //                tablelist += database + "." + tableName + "_" + i + ",";
        //            }
        //            tablelist = tablelist.substring(0, tablelist.length() -1);
        //        } else {
        //            tablelist += database + "." + tableName;
        //        }
        //
        //        System.out.println(tablelist);
    }
}
